package com.sjc.zeye.product.function;

import com.sjc.zeye.product.domain.UserBehavior;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.watermark.Watermark;

import javax.annotation.Nullable;

public class ProductEventTimeExtractor implements AssignerWithPeriodicWatermarks<UserBehavior> {
    private long currentMaxEventTime = 0L;
    private long maxOutputOrderness = 10000;

    // 获取watermark
    @Nullable
    @Override
    public Watermark getCurrentWatermark() {
        return new Watermark(currentMaxEventTime - maxOutputOrderness);
    }

    @Override
    public long extractTimestamp(UserBehavior userBehavior, long l) {
        currentMaxEventTime = Math.max(userBehavior.getTimestamp(),currentMaxEventTime);
        return userBehavior.getTimestamp() * 1000; // 原始数据 精确到秒，但是我们需要精确到毫秒
    }
}
